package com.shujia.dim

import com.shujia.common.SparkTool
import com.shujia.common.utils.poly.Polygon
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object DimScenicGrid extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._
    // 分别加载景区边界表、网格配置表
    val scenicDF: DataFrame = spark.table("dim.dim_scenic_boundary")
    val gridDF: DataFrame = spark.table("dim.dim_geotag_grid")

    val isInBoundaryWithLgLat: UserDefinedFunction = udf((boundary: String, longi: Double, lati: Double) => {
      new Polygon(boundary).contains(longi, lati)
    })

    // 将景区的边界转换成一个一个的网格id
    // 同样没有直接的关联条件，也需要做笛卡尔积
    // 但是由于两份数据都是基本不变的，故做一次笛卡尔积之后就可以多次使用
    // 避免在景区游客表计算中每天定时调度，相当于每天都要做一次笛卡尔积
    scenicDF
      .join(gridDF)
      // 过滤出属于景区内的网格
      .where(isInBoundaryWithLgLat($"boundary", $"center_longi", $"center_lati"))
      .groupBy($"scenic_id", $"scenic_name")
      .agg(concat_ws("|", collect_set($"grid_id")) as "grids")
      .write
      .format("csv")
      .option("sep", ",")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dim/dim_scenic_grid")


  }
}
